package com.song.stormdemo1;

import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;
import backtype.storm.utils.Utils;

public class CountBolt extends BaseRichBolt{  
	  
    private Map stormConf;  
    private TopologyContext context;  
    private OutputCollector collector;  
    public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) {  
        this.stormConf = stormConf;  
        this.context = context;  
        this.collector = collector;  
    }  
  
    private HashMap<String, Integer> map = new  HashMap<String, Integer>();  
    public void execute(Tuple input) {  
        String word = input.getStringByField("word");  
        System.err.println(word);  
        Integer value = map.get(word);  
        if(value==null){  
            value = 0;  
        }  
        value++;  
        map.put(word, value);  
        //把结果写出去  
        System.err.println("============================================");  
        Utils.sleep(2000);  
        for (Entry<String, Integer> entry : map.entrySet()) {  
            System.out.println(entry);  
        }  
    }  
  
    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
          
    }  
      
}  